
package SSF.OS.TCP;

/*
 * tcpSession.java
 *
 * Created: Mon Jul 19 02:24:52 1999 hongbol
 * Revised Sat Nov 20 1999 ato
 * Modified by hongbol 8/16/2000
 * Revised 5/26/01 ato
 * @version 1.3
 */

import SSF.OS.*;
import SSF.Net.Util.*;
import SSF.OS.Socket.*;
import java.util.*;

/**
 * This class represents a TCP session of one end of the connection which is
 * identified by source and destination IP address and port number. It keeps
 * all the state information and controls data transfer by using class 
 * Send_WndManager and Rcv_WndManager.
 */
public class tcpSession extends ProtocolSession{

  /** Option used to switch on/off verbose session debugging output. */
  public boolean DEBUG;

  /** Option used to switch on/off verbose session information output. */
  public boolean INFO;

  /** TCP master service that owns this TCP session. */
  public tcpSessionMaster  tcpServ;

  /** Local (source) port number. */
  int srcPort;

  /** Remote (destination) port number. */
  int destPort;

  /** flow ID for this connection - used by Instrumentation */
  int flowID;

  /** local IP address for this connection - used by Instrumentation */
  int srcIP;

  /** CLOSE Timer basis, default value 0.5 seconds. */
  public static final long close_timer_basis = SSF.Net.Net.seconds(0.5);

  /**  IP header template of this session. */
  public IpHeader ipHdr;

  /** TCP header template of this session. */
  public TCP_Header tcpHdr;

  /** Local IP Session in the ProtocolGraph. */
  public ProtocolSession ipsess;

  /**
   * For listening server TCP session, it is the reference to master server 
   * socket. For other TCP sessions, it is the socket connected to 
   * this TCP session.
   */ 
  public tcpSocket mySocket;

  /** TCP connection state variable. */
  public int tcp_state;

  /** Object managing the send queue and processing the received ACKs */
  public Send_WndManager snd;

  /* Object managing the receive buffer and sending the ACKs. */
  Rcv_WndManager rcv;

  /* Initial sequence number for this TCP connection */
  long ISS;
  
  /* Flag for the delayed ack option. */
  boolean DELAYED_ACK = false;

  boolean ACK_NOW = true;

  /* Protocol status concerning read and write states. */
  int protocol_status; // 1. idle, 2: read, 3: write, 4: ready

  /* session/flow information*/
  String session_info;
    
  Instrumentation INSTR = null;

  /* connection state */
  public final static int CLOSED = 0;
  public final static int LISTEN = 1;
  public final static int SYN_SENT = 2;
  public final static int SYN_RECEIVED = 3;
  public final static int ESTABLISHED = 4;
  public final static int CLOSE_WAIT = 5;
  public final static int CLOSING = 6;
  public final static int FIN_WAIT_1 = 7;
  public final static int FIN_WAIT_2 = 8;
  public final static int LAST_ACK = 9;
  public final static int TIME_WAIT = 10;

  /** maximum segment lifetime in slow ticks*/
  public static int MSL;

  /** maximum FIN_WAIT_2 idle time*/
  public static long maxIdleTime;

  /** Idle time counter */
  int idleTime;

  /** 2 MSL timeout counter */
  int mslTimerCnt;

  /** connection timer */
  int conTimerCnt;

  long duration = 0;

  public String stateToString(int state) {
    String str;
    switch(state) {
      case CLOSED:
        str = "CLOSED";
        break;
      case LISTEN:
        str = "LISTEN";
        break;
      case SYN_SENT:
        str = "SYN_SENT";
        break;
      case SYN_RECEIVED:
        str = "SYN_RECEIVED";
        break;
      case ESTABLISHED:
        str = "ESTABLISHED";
        break;
      case CLOSE_WAIT:
        str = "CLOSE_WAIT";
        break;
      case CLOSING:
        str = "CLOSING";
        break;
      case FIN_WAIT_1:
        str = "FIN_WAIT_1";
        break;
      case FIN_WAIT_2:
        str = "FIN_WAIT_2";
        break;
      case LAST_ACK:
        str = "LAST_ACK";
        break;
      case TIME_WAIT:
        str = "TIME_WAIT";
        break;
      default:
        str = "UNKNOWN TCP STATE";
        break;
    }
     return str;
  }

  /************************* Constructors ************************/

  /** False zero-argument constructor to flag improper direct use of 
    * the tcpSession protocol from configuration files.  
    */
  public tcpSession() throws TCPException {
    throw new TCPException("Use tcpSessionMaster instead of tcpSession.");
  }

  /**
   * Initialize TCP session with LISTEN state and port number of the listening
   * server application
   */
  public tcpSession(tcpSessionMaster serv)
  {
    tcpServ = serv;
  }

  /**
   * Initialize TCP session with SYN_RECEIVED state and IP address 
   * and port number of source and destination
   */
  public tcpSession(tcpSessionMaster serv, tcpSocket upSession, 
                    ProtocolSession lowSession, int SrcIpAddr, int SrcPort, 
                    int DesIpAddr, int DesPort)
  {
    tcpServ = serv;
    MSL = (int)(serv.MSL/serv.TCP_SLOW_INTERVAL);
    maxIdleTime = serv.MaxIdleTime;
    mySocket = upSession;
    //flowID = mySocket.sockNo;
    flowID = tcpServ.getSessionID();
    ipsess = lowSession;
    srcPort = SrcPort;
    destPort = DesPort;
    tcpHdr = new TCP_Header(SrcPort, DesPort,  tcpServ.ISS, 0, 0, (byte)0);
    ipHdr = new IpHeader(Protocols.TCP_PRTL_NUM, SrcIpAddr, DesIpAddr);

    ipHdr.carryPayload(tcpHdr);
    ISS = tcpServ.ISS;
    DEBUG = tcpServ.DEBUG;
    INFO = tcpServ.INFO;

    // connection timer is set for 75 sec to allow 2 retransmissions of SYN
    conTimerCnt = (int)(SSF.Net.Net.seconds(75)/
			(double)tcpServ.TCP_SLOW_INTERVAL);

    session_info = "TCP host " + ((SSF.Net.Host)tcpServ.inGraph()).nhi
      + " src={" + SSF.Net.Util.IP_s.IPtoString(tcpServ.ipAddr)
      + ":" + srcPort + "} dest={" 
      + SSF.Net.Util.IP_s.IPtoString(ipHdr.DEST_IP) 
      + ":" + destPort +"} ";
  }

  /************************* Class Methods ***********************/

  String sessionInfo(String str){
    return (tcpServ.inGraph().now()/(double)SSF.Net.Net.seconds(1.0)
	    + " " + session_info + str);
  }

  /** Active open a connection by sending a message with a SYN flag.
   */
  public void ActiveOpen() {
    TCP_Header tcp_h = new TCP_Header(srcPort, destPort, ISS, 0, 0, (byte)0);
    tcp_h.flags = TCP_Header.SYN;
    snd = new Send_WndManager(this,tcpServ);
    rcv = new Rcv_WndManager(this, tcpServ);

    if(DEBUG) {
      INSTR = new Instrumentation(tcpServ, this);
      INSTR.takeSnapshot(getSimTime(),INSTR.EVT_OPEN);
    }

    TCP_Message message = new TCP_Message(tcp_h, ipHdr.SOURCE_IP,
                                          ipHdr.DEST_IP, 0);
    snd.SendBuffer.append(message,0,tcp_h,ipHdr.SOURCE_IP,
			  ipHdr.DEST_IP,false);
    snd.Send_Data_Process();
    tcp_state = SYN_SENT;
    if(INFO || DEBUG)
      System.out.println(sessionInfo(" Active Open"));

  }

  /**
   * Passive open a listening connection.
   */
  public void PassiveOpen(){
    tcp_state = LISTEN;
  }

  /**
   * State transfer when FIN is received.
   */
  public void CloseStateTransfer1(){
    switch (tcp_state){
    case ESTABLISHED:
      tcp_state = CLOSE_WAIT;
      break;
    case FIN_WAIT_1:
      tcp_state = CLOSING;
      break;
    case FIN_WAIT_2:
      tcp_state = TIME_WAIT;
      if(mslTimerCnt == 0) // set 2 MSL timer
        mslTimerCnt = 2 * MSL;
      break;
    default:
      break;
    }
    if(DEBUG)
      INSTR.takeSnapshot(getSimTime(), INSTR.EVT_STATE);
  }

  /**
   * State transfer when
   * 1) FIN is sent;
   * 2) ACK of FIN is received;
   *
   */
  public void CloseStateTransfer2(){
    switch (tcp_state){
      case ESTABLISHED:
        tcp_state = FIN_WAIT_1;
        break;

      case CLOSE_WAIT:
        tcp_state = LAST_ACK;
        break;

      case LAST_ACK:
        tcp_state = CLOSED;
        mySocket.disconnected();
        tcpServ.removeConnect(this); // remove connection
        break;

      case FIN_WAIT_1:
        tcp_state = FIN_WAIT_2;
        if(mslTimerCnt == 0) // set FIN_WAIT_2 timer
          mslTimerCnt = 2 * MSL;
        break;

      case CLOSING:
        tcp_state = TIME_WAIT;
        if(mslTimerCnt == 0) //set 2MSL timer
          mslTimerCnt = 2 * MSL;
        break;

      default:
        break;
    }
    if(DEBUG)
      INSTR.takeSnapshot(getSimTime(), INSTR.EVT_STATE);
  }

  
  public void CloseStateTransfer3(){
    switch (tcp_state){
    case FIN_WAIT_1:
      tcp_state = TIME_WAIT;
      if(mslTimerCnt == 0)
        mslTimerCnt = 2 * MSL;
      
      break; 
    default :
      break; 
    }
    if(DEBUG)
      INSTR.takeSnapshot(getSimTime(), INSTR.EVT_STATE);
  }
  
  /** slow timer function called every 500 ms */
  public void slowTimeout(){
    /* connection timer is timeout( 75 sec )*/
      if((conTimerCnt != 0) && (--conTimerCnt == 0)){
	snd.rexTimerCnt = 0;   //cancel retransmission timer
	drop(socketMaster.ETIMEDOUT);
      }
      if((snd.rexTimerCnt != 0) && (--snd.rexTimerCnt == 0)){
        if(DEBUG && (tcpServ.rtoCountWr != null)) {
          float etime = getSimTime();
          float stime = snd.SendBuffer.dummyItem.next.getPrevSendTime();
          tcpServ.INSTR.timeoutRecord(stime,etime,flowID, 
                                      snd.tickToSeconds(snd.rexTimeout));
        }
        snd.Rexmit();
      }
      else{
        if(DEBUG) snd.dump(snd.RTO_DUMP, null);
      }
      if((mslTimerCnt != 0) && (--mslTimerCnt == 0)){
        TwoMSLtimeout();
      }
      idleTime++;
      duration++;
      if(snd.rtt > 0)snd.rtt++;
      if(DEBUG)
        INSTR.takeSnapshot(getSimTime(), INSTR.EVT_SLOWTICK);
  }

  /** fast timer function called every 200 ms */
  public void fastTimeout(){

    if(DELAYED_ACK & ACK_NOW) {
      rcv.Send_Ack_Process();
      DELAYED_ACK = false;
      ACK_NOW = false;
    }
    if(DEBUG){
      INSTR.takeSnapshot(getSimTime(), INSTR.EVT_FASTTICK);
    }
    
  }

  /**
   * Performs 2MSL timer and FIN_WAIT_2 timer timeout functions.
   */
  public void TwoMSLtimeout(){
    if((tcp_state != TIME_WAIT) //Is FIN_WAIT_2 timer
       && (idleTime <= maxIdleTime))
      mslTimerCnt = 2 * MSL;
    else {    
      if(INFO || DEBUG){
        if(tcp_state == TIME_WAIT)
          System.out.println(
                     sessionInfo(" 2MSL timeout, connection closed"));
        else
          System.out.println(sessionInfo
                             (" FIN_WAIT_2 timeout, connection closed"));
      }
      tcp_state = CLOSED;
      mySocket.disconnected();
      tcpServ.removeConnect(this);// remove connection
    } 
  }

  /**
   * Tell the caller if the TCP connection exists
   * @return true if the state is  ESTABLISHED
   */
  
  public boolean opened(){
    return tcp_state >= ESTABLISHED;
  }

  /**
   * Tell the caller if the TCP connection exist
   * @return true if the state is CLOSED
   */

  public boolean closed(){
    return tcp_state == CLOSED;
  }
  
  /**
   * Get the value of protocol_status.
   * @return Value of protocol_status.
   */
  public int  get_protocol_status() {return protocol_status;}
  
  /**
   * Set the value of protocol_status.
   * @param v  Value to assign to protocol_status.
   */
  public void set_protocol_status(int   v) {this.protocol_status = v;}

  /** 
   * drop a TCP session with error
   */
  void drop(int errorno){

    tcp_state = CLOSED;

    //return error number to application
    mySocket.failure(errorno);
    tcpServ.removeConnect(this);  // remove connection
  }

  /** 
   * send reset to peer connection, not used.
   */
  void sendReset(){

    tcpHdr.flags = TCP_Header.RST;
    TCP_Message msg = new TCP_Message(tcpHdr, ipHdr.SOURCE_IP, ipHdr.DEST_IP,
				      0);
    snd.SendBuffer.append(msg,0,tcpHdr,ipHdr.SOURCE_IP,ipHdr.DEST_IP,
			  false);
    snd.Send_Data_Process();
  }

  /** 
   * Socket closes its TCP session.
   * @return true if success, false if failed
   */
  public boolean close() throws TCPException{
    if((tcp_state != CLOSE_WAIT) && (tcp_state != ESTABLISHED)){
      throw new TCPException("error: close() in state "
                             +stateToString(tcp_state));
    }

    tcpHdr.flags = TCP_Header.FIN;
    TCP_Message message = new TCP_Message(tcpHdr, ipHdr.SOURCE_IP,
                                          ipHdr.DEST_IP, 0);
    if(INFO)
      System.out.println(sessionInfo(" Active Close"));
    if (INFO && (tcp_state == CLOSE_WAIT))
      System.out.println(sessionInfo(" Passive Close"));
    
    snd.SendBuffer.append(message, 0, tcpHdr, ipHdr.SOURCE_IP,
                          ipHdr.DEST_IP, false);
    snd.FINsent = true;
    snd.Send_Data_Process();
    if(DEBUG){
      INSTR.takeSnapshot(getSimTime(),INSTR.EVT_CLOSE);
    }
    return true;
  }

  public float getSimTime(){
    return tcpServ.inGraph().now()/(float)SSF.Net.Net.seconds(1);
  }

  /**
   * Classify and process the incoming TCP message according to message
   * flags and connection state.
   * @param message incoming TCP message
   * @param fromSession ProtocolSession that invoked the push() method,
   * here IP.
   * @return result of the push action: If successful return true, otherwise
   * return false.
   */
  public boolean push(ProtocolMessage message, ProtocolSession fromSession)
         throws ProtocolException{

    //If data come from IP session, process it at once
    if(fromSession instanceof IP){
      TCP_Message tcpMsg = (TCP_Message)message;

      //System.out.println(sessionInfo("state= "+tcp_state+" rcvd flag: "+tcpMsg.flags
      //                                +" ackNo= "+tcpMsg.ACKno));

      if(snd != null) snd.checkIdle();
      idleTime = 0;

      /*
	When receive RST:
	SYN_RCVD: new socket is dropped. back to LISTEN
                     if with ACK, error code CONNREFUSED
                     if no ACK, just drop with no error code
	ESTABLISH FIN1 FIN2 CLOSE_WAIT: drop with error code CONNRESET
        after CLOSE_WAIT, RST means full close and no more data can be sent
                     just drop, no error code
      */
      if((tcpMsg.flags & TCP_Header.RST) != 0){ 
	switch(tcp_state){
        case SYN_RECEIVED:
	  drop(socketMaster.ECONNREFUSED);
	  break;
        case ESTABLISHED:
        case FIN_WAIT_1:
        case FIN_WAIT_2:
        case CLOSE_WAIT:
	  drop(socketMaster.ECONNRESET);
	  break;
        case CLOSING:
        case LAST_ACK:
        case TIME_WAIT:
	  drop(0);
	  break;
        }
      }	
      
      /*
	if ACK is received in LISTEN state, ACK is not for the SYN in 
	SYN_SENT & SYN_RECEIVED state, send a RST to peer connection
      */
      if((tcpMsg.flags & TCP_Header.ACK) != 0){
	switch(tcp_state){
	  case SYN_SENT:
	  case SYN_RECEIVED:
	    if((tcpMsg.ACKno <= ISS)||(tcpMsg.ACKno > (snd.LastSent + 1))) {
	
	    } else {
	      break;
	    }
	
	  case LISTEN:
	    tcpServ.respond(ipHdr.SOURCE_IP, ipHdr.DEST_IP, srcPort, destPort,
			    tcpServ.RcvWndSize*tcpServ.MSS, tcpMsg.ACKno,
			    tcpMsg.SEQno+tcpMsg.seglen,
			    (byte)(TCP_Header.RST|TCP_Header.ACK), 0);	
	    break;
	
	  default:
	    break;
	}
      }

      switch(tcp_state) {
        case CLOSED:
          break;

        case LISTEN:
          if(tcpMsg.flags == TCP_Header.SYN) {
            snd = new Send_WndManager(this, tcpServ);
            rcv = new Rcv_WndManager(this, tcpServ);

            // tcpMsg fields changed into reply message
            srcPort = tcpMsg.DEST_port;
            destPort = tcpMsg.SOURCE_port;
            if(DEBUG)
              INSTR.takeSnapshot(getSimTime(), INSTR.EVT_PKRCV);

            rcv.Rcv_WndInit(tcpMsg);  // sends SYN ACK
            tcp_state = SYN_RECEIVED;

            if(DEBUG)
              INSTR.takeSnapshot(getSimTime(), INSTR.EVT_PKSND);
            if(INFO)
              System.out.println(sessionInfo("SYN recvd"));
            return true;
          }
          break;
        
        case SYN_SENT:
	  // when RST received and the ACKno is correct, it means connection
	  // is refused by server
	  if((tcpMsg.flags & TCP_Header.RST) != 0){
	    if(((tcpMsg.flags & TCP_Header.ACK) != 0)
	         && (tcpMsg.ACKno > snd.FirstUnACKed)) {
	      drop(socketMaster.ECONNREFUSED);
	    }else {
	      drop(0);
	    }
	    break;
	  }
	
	  // received SYN-ACK
	  if(((tcpMsg.flags & TCP_Header.ACK) != 0)
	       &&((tcpMsg.flags & TCP_Header.SYN) != 0)){
            rcv.Rcv_WndInit(tcpMsg);
            snd.Rcv_Ack_Process(tcpMsg);
            if(DEBUG)
              INSTR.takeSnapshot(getSimTime(), INSTR.EVT_PKRCV);

            tcp_state = ESTABLISHED;

	    // Cancel connection Timer
	    conTimerCnt = 0;

            //Notify the established state
            mySocket.connected();
            snd.sub_state = 1;
            snd.Send_Data_Process();
            if(DEBUG)
              INSTR.takeSnapshot(getSimTime(), INSTR.EVT_PKSND);
            return true;
          }

        case SYN_RECEIVED:

          if((tcpMsg.flags & TCP_Header.ACK) != 0){
            snd.Rcv_Ack_Process(tcpMsg);
            if(DEBUG)
              INSTR.takeSnapshot(getSimTime(), INSTR.EVT_PKRCV);
            tcp_state= ESTABLISHED;

	    // Cancel connection Timer
	    conTimerCnt = 0;

            //Notify socket that connection is established
            mySocket.listeningSocket.addConnection(mySocket);
            if(tcpMsg.seglen > 0){
              try {
                rcv.Rcv_DataProcess(tcpMsg);
              } catch (TCPException e) {
                System.err.println(sessionInfo("Receive packet overflow! ") + e);
                e.printStackTrace();
              }
            }
            snd.sub_state = 1;
            snd.Send_Data_Process();
            if(DEBUG)
              INSTR.takeSnapshot(getSimTime(), INSTR.EVT_PKSND);
	    return true;
          }

        case ESTABLISHED:
        case FIN_WAIT_1:
        case FIN_WAIT_2:
          if((tcpMsg.seglen > 0) || ((tcpMsg.flags & tcpMsg.FIN) != 0))
            try{
              rcv.Rcv_DataProcess(message);
            }catch(TCPException e){
              System.err.println(sessionInfo("Receive packet overflow! ") + e);
              e.printStackTrace();
            }

          if((tcpMsg.flags & TCP_Header.ACK) != 0)
            snd.Rcv_Ack_Process(tcpMsg);
          if(DEBUG)
            INSTR.takeSnapshot(getSimTime(), INSTR.EVT_PKRCV);

          return true;

        /*
        case FIN_WAIT_2:  moved above by ato 5/28/01
          if ((tcpMsg.flags & TCP_Header.ACK) != 0)   // <-- problem?
            return false;
          else {
            try {
              rcv.Rcv_DataProcess(message);
            } catch (TCPException e) {
              System.err.println(sessionInfo("Receive packet overflow! ") + e);
              e.printStackTrace();
            }
          }
          if(DEBUG)
            INSTR.takeSnapshot(getSimTime(), INSTR.EVT_PKRCV);
          break;
        */

        case CLOSE_WAIT:
        case CLOSING:
        case LAST_ACK:
          if((tcpMsg.flags & TCP_Header.ACK) != 0)
            snd.Rcv_Ack_Process(tcpMsg);

          if(DEBUG)
            INSTR.takeSnapshot(getSimTime(), INSTR.EVT_PKRCV);
          return true;
      
        /* Have stopped to send data, can receive data, half closed */
        case TIME_WAIT:
          if(DEBUG){
            System.err.print(sessionInfo("**Warning** Message received at TIME_WAIT"));
            tcpMsg.printMsg();
          }
          break;

        default:
          break;
      }
    }
    return false;
  }

  /** test if the connection is closed
   */
  public boolean isClosed(){
    return tcp_state == CLOSED;
  }

  /********************** tcpSocket interface ************************/

  // only one blocking read() or write() can be in progress at any time

  Object[] readObj = new Object[1];
  boolean isReadingObject = false;
  int readSize;
  int dataReceived = 0;
  int writeSize;
  dataMessage writeMsg;
  Continuation readCont = null;
  Continuation writeCont = null;

  Continuation app_call_waiting;

  // change ato 5/28/01: write returns success when all written bytes
  // are acked. Previous versions returned success when send buffer was drained.

  public void write(dataMessage dmsg, Continuation caller){
    if (writeCont != null){
      System.err.println(sessionInfo("write error - socket already in blocking write"));
      caller.failure(socketMaster.EBUSY);
      return;
    }
    writeMsg = dmsg;
    writeCont = caller;
    writeSize = dmsg.size;
    dataFeed();
    if(DEBUG)
      INSTR.takeSnapshot(getSimTime(), INSTR.EVT_SOCKWRITE);
  }

  // callback from Send_WndManager.Rcv_Ack_Process each time
  // SendBuffer has enough free space to accomodate more data.
  // Note that writeMsg.size is decremented, and should not
  // be accessed by read().

  void dataFeed(){
    if(writeCont == null)
      return;
    tcpHdr.flags = 0;
    int size = snd.SendBuffer.append(writeMsg, writeMsg.size, tcpHdr,
                                     ipHdr.SOURCE_IP, ipHdr.DEST_IP, false);
    if(DEBUG)
      INSTR.takeSnapshot(getSimTime(), INSTR.EVT_PKSND);

    snd.Send_Data_Process();
    writeMsg.size -= size;
  }

  // Callback from Send_WndManager.Rcv_Ack_Process() when
  // a segment was acked. added ato 5/28/01.

  void dataAcked(int nbytes) {
    if(writeCont == null) return;
    //System.err.println(sessionInfo("dataAcked: "+nbytes));
    writeSize -= nbytes;

    // return success when all data were acked
    if(writeSize <= 0){
      Continuation app_call_waiting = writeCont;
      writeCont = null;
      mySocket.appCallWaiting = false;
      //mySocket.socketInfo("wrote data");
      app_call_waiting.success();
    }
  }

  // change ato 5/28/01: allow for data arrival before a read().
  // Previous versions assumed that read() is always called before
  // any data arrive.

  // obj[0] is a handle to the application's object passed through
  // the read() method of the socket.

  public void read(Object[] obj, int nbytes,  Continuation caller){
    //System.err.println(sessionInfo("read obj= "+obj+" nbytes= "+nbytes));
    if(DEBUG)
      INSTR.takeSnapshot(getSimTime(), INSTR.EVT_SOCKREAD);

    if (readCont != null){
      System.err.println(sessionInfo("read error - socket already in blocking read"));
      caller.failure(socketMaster.EBUSY);
      return;
    }

    // if requested data is in the buffer, get it from buffer
    // We use the fact that a data object arrives in the first TCP segment.

    if(nbytes <= dataReceived) {
      dataReceived -= nbytes;

      if(obj != null) {
        if (objBuffer.size() == 0) {
          System.err.println(
            sessionInfo("ERROR: TCP socket executing read(Object,int,Continuation)"+
                        " but only virtual data arrived, no Object"));
          System.exit(-1);
        }
        obj[0] = objBuffer.get(0);
        objBuffer.remove(0);
      }
      caller.success();
    } else {
      readObj = obj;
      if(readObj != null)
        isReadingObject = true;
      readCont = caller;
      readSize = nbytes;
    }
  }

  // Callback from RCV_WndManager.ReassembleMessage() (directly and via
  // RCV_WndManager.push_Process()) each time new in-sequence
  // data has arrived.
  // Note that dataCollect() may be called before the read(), thus
  // we need to cache the data until read() is called.

  private ArrayList objBuffer = new ArrayList(1);

  void dataCollect(dataMessage dmsg, int size){
    if(dmsg != null)
        objBuffer.add(dmsg.data);

    dataReceived += size;
    //System.err.println(sessionInfo("dataReceive: obj= "+dmsg+" size= "+size+
    //                               ", waiting for= "+(readSize - dataReceived)));

    // We use the fact that a data object arrives in the first TCP segment.
    if((readCont != null) && (dataReceived >= readSize)) {
      if(isReadingObject) {
        if (objBuffer.size() == 0) {
          System.err.println(
            sessionInfo("ERROR: TCP socket executing read(Object,int,Continuation)"+
                        " but only virtual data arrived, no Object"));
          System.exit(-1);
        }
        readObj[0] = objBuffer.get(0);
        objBuffer.remove(0);
        isReadingObject = false;
      }
      Continuation app_call_waiting = readCont;
      readCont = null;
      mySocket.appCallWaiting = false;
      dataReceived -= readSize;
      app_call_waiting.success();
    }
  }
    
}

